from datetime import datetime
from typing import Literal
from collections import Counter
import numpy
import pandas
import time
from dateutil.relativedelta import relativedelta


class PrintTimestamp:
    def __init__(self):
        self.print_timestamp_str = ""


    def print_timestamp(self, text:str):
        self.print_timestamp_str = text
        print(f"{time.strftime('%Y/%m/%d %H:%M:%S')}\n\t" + text +"\n")




class BasicFunction(PrintTimestamp):
    def __init__(self,df:pandas.DataFrame, numeric_columns:list=None, datetime_columns:list=None):
        super().__init__()

        df_col_set = set(df.columns.tolist())
        if numeric_columns is not None:
            for nc_k in numeric_columns:
                df[nc_k] = df[nc_k].replace(numpy.nan, 0).replace(pandas.NA, 0).replace("", 0)

            df_col_set -= set(numeric_columns)

        for dcs_k in df_col_set:
            df[dcs_k] = df[dcs_k].replace(numpy.nan, None).replace(pandas.NA, None).replace("", None)


        self.df = df
        # self.table_name = table_name



    def transform_data_type(
            self,
            to_datetime_list:list = None,
            to_numeric_list:list = None
    ):
        """
        转变列数据的类型。
        :param to_datetime_list:填入需要转变为日期类型的列名称
        :param to_numeric_list:填入需要转变为数值类型的列名称
        :return:
        """

        if to_datetime_list is not None:
            for k in to_datetime_list:
                try:
                    self.df[k] = pandas.to_datetime(self.df[k], errors='raise', yearfirst=True,
                                                    format='%Y-%m-%d %H:%M:%S')

                except ValueError:
                    try:
                        self.df[k] = pandas.to_datetime(self.df[k], errors='raise', yearfirst=True,
                                                        format='%Y/%m/%d')
                    except ValueError as e:
                        try:
                            self.df[k] = pandas.to_datetime(self.df[k], errors='raise', yearfirst=True,
                                                            format='%Y-%m-%d')
                        except ValueError as e:
                            raise ValueError(
                                f"【{k}】列没有成功转为日期格式！！！\n{e}"
                            )
                except KeyError as e:
                    raise KeyError(
                         f"【{k}】列不存在！！！"
                    )

        if to_numeric_list is not None:
            for j in to_numeric_list:
                try:
                    self.df[j] = pandas.to_numeric(self.df[j], errors='raise')
                except ValueError as e:
                    raise ValueError(
                        f"【{j}】列没有成功转为数值格式！！！\n{e}"
                    )
                except KeyError as e:
                    raise KeyError(
                        f"【{j}】列不存在！！！"
                    )

        return self





    def rename_columns_on_df(self, transform_dict:dict):
        """
        重命名列名。
        :param transform_dict:
        :return:
        """

        self.df.rename(
            columns = transform_dict,
            inplace = True,
            errors = "raise"
        )

        return self




    def select_columns_on_df(self, columns:list):
        """
        选取需要保留的列
        :param columns:
        :return:
        """

        self.df = self.df.loc[:, columns]

        return self




    def del_columns_on_df(self, columns: list):
        """
        选取需要删除的列
        :param columns:
        :return:
        """

        self.df.drop(columns=columns, inplace=True)

        return self




    def text_trim_on_df(self,all_number_col:list = None):
        """
        清除值前后的空白字符。如果指定了包含全数值文本字符串的列，则还会清除前后的英文单引号字符。
        :param all_number_col:
        :return:
        """
        other_col = list(set(self.df.columns.tolist()) - set(all_number_col))

        for ntc_k in all_number_col:
            self.df[ntc_k] = self.df[ntc_k].apply(
                lambda x: (x if x is None else x.strip("' "))
            )

        for oc_k in other_col:
            self.df[oc_k] = self.df[oc_k].apply(
                lambda x: (x if x is None else x.strip(" "))
            )

        return self




    def select_rows_on_df(self, select_text:str, s_local_dict:dict = None):
        """
        使用文本筛选dataframe记录行。例："列名 != '你好'"
        :param s_local_dict: 添加局部变量
        :param select_text:
        :return:
        """

        self.df.query(expr=select_text, local_dict = s_local_dict, inplace=True)

        return self




    def check_missing_value_on_df(self, check_cols:list, key_cols:list):
        """
        检查列是否有缺失值。若有则报错。
        """
        missing_values = {None, ""}
        keys_cols_sep = '_'

        for cc_k in check_cols:
            if len(missing_values & set(self.df[cc_k])) > 0:
                question_df = self.df.query(f"{cc_k} in @missing_values").loc[:, key_cols]

                question_df.drop_duplicates(inplace=True)
                question_df["索引"] = question_df.apply(
                    lambda s: s.str.cat(sep=keys_cols_sep),
                    axis=1
                )

                raise ValueError(
                    f"【{cc_k}】列包含有缺失值。\n"
                    + f"涉及的关键列【{keys_cols_sep.join(key_cols)}】的索引值为：{question_df["索引"].str.cat(sep="、")}"
                )

        return self


    def check_value_on_df(
            self,
            check_cols_name:list,
            check_values:list,
            is_include:bool,
            key_cols:list
    ):
        """
        检查是否仅包含或完全不包含某些值。异常则报错。
        :param check_cols_name:
        :param check_values:
        :param is_include:
        :param key_cols:
        :return:
        """
        def __is_all_include(col_name:str):
            return check_values >= set(self.df[col_name])

        def __is_not_include(col_name:str):
            return len(check_values & set(self.df[col_name])) == 0

        def __select_by_is_all_include(col_name: str):
            return f"{col_name} not in @check_values"

        def __select_by_is_not_include(col_name: str):
            return f"{col_name} in @check_values"


        check_values = set(check_values)
        keys_cols_sep = '_'


        # 根据is_include选择检查方法和查询方式
        if is_include:
            __include_fun = __is_all_include
            __select = __select_by_is_all_include
        else:
            __include_fun = __is_not_include
            __select = __select_by_is_not_include


        for ccn_k in check_cols_name:
            if not __include_fun(ccn_k):
                question_df = self.df.query(__select(ccn_k))

                exception_values = '、'.join([str(lcv_k) for lcv_k in list(set(question_df[ccn_k]))])

                question_df = question_df.loc[:, key_cols].drop_duplicates()
                question_df["索引"] = question_df.apply(
                    lambda s: s.str.cat(sep = keys_cols_sep),
                    axis = 1
                )

                raise ValueError(
                    f"【{ccn_k}】列中不应包含这些值：{exception_values}\n"
                    + f"涉及的关键列【{keys_cols_sep.join(key_cols)}】的索引值为：{question_df["索引"].str.cat(sep="、")}"
                )

        return self



    def check_last_month_max_data_on_datetime_col(self, data_col_name:list):
        """
        检查日期类型列中每月的日期是否大于25日。
        :return:
        """
        for dcn_k in data_col_name:
            temp_df = self.df.loc[:, [dcn_k]]
            temp_df["所属年月"] = temp_df[dcn_k].apply(
                lambda x: x.strftime("%Y年%m月")
            )

            # 仅筛选出上个月的日期
            temp_df = temp_df[temp_df["所属年月"] == (datetime.now() + relativedelta(months=-1,day=1)).strftime("%Y年%m月")]

            g_t_df = temp_df.groupby(by=["所属年月"],as_index=False)[dcn_k].max()
            g_t_df["日"] = g_t_df[dcn_k].apply(
                lambda x: x.day
            )

            if g_t_df["日"].min() <= 25:
                raise ValueError(
                    f"【{dcn_k}】列存在“上个月的最大日期小于或等于25日”的情况。"
                )

        return self



    def transform_month_start_date_on_datetime_col(self, data_col_name:list):
        """
        将日期列的值转换为对应的月份开始值
        :return:
        """
        for dcn_k in data_col_name:
            self.df[dcn_k] = self.df[dcn_k].apply(
                lambda x: datetime(year=x.year, month=x.month, day=1)
            )

        return self



    def drop_all_na_row(self):
        """
        删除掉空行。
        :return:
        """
        self.df = self.df[~self.df.isna().all(axis=1)]



    def is_equals(self, other_df:pandas.DataFrame, print_error:bool = False):
        """
        两数据表相互比较
        :param print_error:
        :param other_df:
        :return:
        """
        def __find_different_by_list(s_list, o_list, get_key_or_index:Literal["key", "index"]):
            """
            先用集合判断是否一致，若否则是个数上不一致，若是则是存在重复值
            :param s_list:
            :param o_list:
            :param get_key_or_index:
            :return:
            """
            if get_key_or_index == "key":
                s_count_df = pandas.DataFrame(data=Counter(s_list),index=[0]).T.rename(
                    columns={0:"计数"}).reset_index(names=["元素值"])
                o_count_df = pandas.DataFrame(data=Counter(o_list), index=[0]).T.rename(
                    columns={0: "计数"}).reset_index(names=["元素值"])

                result_df = pandas.merge(
                    left=s_count_df,
                    right=o_count_df,
                    how="outer",
                    on="元素值",
                    suffixes=["_s","_o"]
                ).fillna(0)
                result_df["差值"] = result_df["计数_o"] - result_df["计数_s"]

                if set(result_df["差值"]) == {0}:
                    return None
                else:
                    result_df = result_df[result_df["差值"] != 0]
                    result_df["元素及差值"] = result_df[["元素值","差值"]].apply(
                        lambda s: "(" + str(s["元素值"]) + ":" + str(int(abs(s["差值"]))) + ")",
                        axis=1
                    )


                    result_str = ""

                    if result_df[result_df["差值"] > 0].shape[0] == 0:
                        result_str += f"目标列表缺失的元素及个数为：{result_df.loc[result_df["差值"] < 0, "元素及差值"].str.cat(sep="、")}\n\n"
                    else:
                        result_str += f"目标列表新增的元素及个数为：{result_df.loc[result_df["差值"] > 0, "元素及差值"].str.cat(sep="、")}\n\n"

                        if result_df[result_df["差值"] < 0].shape[0] > 0:
                            result_str += f"目标列表缺失的元素及个数为：{result_df.loc[result_df["差值"] < 0, "元素及差值"].str.cat(sep="、")}\n\n"


                    raise Exception(result_str)
            else:
                if len(s_list) != len(o_list):
                    return -1

                if s_list != o_list:
                    for sl_i in range(len(s_list)):
                        if s_list[sl_i] != o_list[sl_i]:
                            return sl_i

                    raise Exception("这是不应该到达的地方。程序有误！！！——1")
                else:
                    return None




        sdct_list = self.df.columns.tolist()
        try:
            __find_different_by_list(sdct_list, other_df.columns.tolist(), get_key_or_index="key")
        except Exception as e:
            if print_error:
                raise Exception(f"发现两数据表的字段列不一致。详情：\n{e}")
            else:
                return False


        temp_sdf = self.df.reset_index(names=["*index*"]).sort_values(by=sdct_list,ignore_index=True)
        other_df = other_df.reset_index(names=["*index*"]).sort_values(by=sdct_list,ignore_index=True)


        # if temp_sdf[sdct_list].equals(other_df[sdct_list]):
        #     return True
        # else:
        for col in sdct_list:
            result_diff = __find_different_by_list(list(temp_sdf[col]), list(other_df[col]), get_key_or_index="index")
            if result_diff is None:
                pass
            elif result_diff == -1:
                if print_error:
                    raise Exception(f"发现两数据表的【{col}】列值个数不一致！！！")
                else:
                    return False
            else:
                if print_error:
                    raise Exception(f"发现‘源数据表’【{col}】列第{temp_sdf["*index*"][result_diff]+2}行值：{temp_sdf[col][result_diff]}\n"
                                    f"与‘目标数据表’【{col}】列第{other_df["*index*"][result_diff]+2}行值：{other_df[col][result_diff]}\n"
                                    "不一致！！！")
                else:
                    return False

        return True




    def get_df(self) -> pandas.DataFrame:
        return self.df





if __name__ == "__main__":
    # count_list = Counter([2,51,5,85,8,5,2,5,8,2,158,5])

    # s = str(datetime(2025,1,12))

    dd = pandas.DataFrame(
        data={
            "a": [1, 2, 1],
            "b": ['A', 'B', 'C']
        }
    )

    ddd = dd.loc[dd["a"] == 1, "b"].str.cat(sep="、")
    tt = ddd

    #
    #
    # dd2 = pandas.DataFrame(
    #     data={
    #         "a": [1, 2, 1],
    #         "b": ['B', 'A', 'D']
    #     }
    # )
    #
    #
    # tt = pandas.merge(
    #     left = dd,
    #     right = dd2,
    #     how = "outer",
    #     on = "b",
    #     suffixes = ("_left", "_right")
    # )





    pass
